package com.jblv.frame.common.kafka.config;



import kafka.consumer.ConsumerConfig;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;

import java.util.Properties;

/**
 * @author happyonion.yangtao
 * @version 1.0
 * @title KafkaConfigUtils
 * @projectName jblv-frame
 * @packageName com.jblv.web
 * @description Kafka配置工具类
 * @date Create at 2020/4/17 13:37
 * @since jdk1.8
 */
public class KafkaConfigUtils {

    public KafkaConfigUtils() {
    }

    public static ProducerConfig getProducerConfig() {
        Properties props = new Properties();
        props.put("zookeeper.connect", KafkaConfigProperties.getValue("zookeeper.connect"));
        props.put("zookeeper.session.timeout.ms", KafkaConfigProperties.getValue("zookeeper.session.timeout.ms"));
        props.put("zookeeper.sync.time.ms", KafkaConfigProperties.getValue("zookeeper.sync.time.ms"));
        //设置分区器
        props.put("partitioner.class", "com.jblv.frame.common.kafka.DefaultPartitioner");
        //boker和分区的静态信息
        props.put("metadata.broker.list", KafkaConfigProperties.getValue("metadata.broker.list"));
        //设置序列化类，可以写类的全路径
        props.put("serializer.class", KafkaConfigProperties.getValue("serializer.class"));
        props.put("key.serializer.class", KafkaConfigProperties.getValue("key.serializer.class"));
        props.put("request.required.acks", KafkaConfigProperties.getValue("request.required.acks"));
        ProducerConfig config = new ProducerConfig(props);
        return config;
    }

    public static ProducerConfig getProducerConfigForString() {
        Properties props = new Properties();
        props.put("zookeeper.connect", KafkaConfigProperties.getValue("zookeeper.connect"));
        props.put("zookeeper.session.timeout.ms", KafkaConfigProperties.getValue("zookeeper.session.timeout.ms"));
        props.put("zookeeper.sync.time.ms", KafkaConfigProperties.getValue("zookeeper.sync.time.ms"));
        props.put("metadata.broker.list", KafkaConfigProperties.getValue("metadata.broker.list"));
        props.put("serializer.class", StringEncoder.class.getName());
        props.put("request.required.acks", KafkaConfigProperties.getValue("request.required.acks"));
        Thread.currentThread().setContextClassLoader(null);
        ProducerConfig config = new ProducerConfig(props);
        return config;
    }

    public static ConsumerConfig getConsumerConfig() {
        Properties props = new Properties();
        props.put("zookeeper.connect", KafkaConfigProperties.getValue("zookeeper.connect"));
        props.put("group.id", KafkaConfigProperties.getValue("group.id"));
        props.put("zookeeper.session.timeout.ms", KafkaConfigProperties.getValue("zookeeper.session.timeout.ms"));
        props.put("zookeeper.sync.time.ms", KafkaConfigProperties.getValue("zookeeper.sync.time.ms"));
        props.put("auto.commit.interval.ms", KafkaConfigProperties.getValue("auto.commit.interval.ms"));
        props.put("auto.offset.reset", KafkaConfigProperties.getValue("auto.offset.reset"));
        props.put("metadata.broker.list", KafkaConfigProperties.getValue("metadata.broker.list"));
        props.put("serializer.class", KafkaConfigProperties.getValue("lisener.serializer.class"));
        props.put("key.serializer.class", KafkaConfigProperties.getValue("key.serializer.class"));
        props.put("rebalance.max.retries", KafkaConfigProperties.getValue("rebalance.max.retries"));
        props.put("rebalance.backoff.ms", KafkaConfigProperties.getValue("rebalance.backoff.ms"));
        ConsumerConfig config = new ConsumerConfig(props);
        return config;
    }
}
